package com.wzh.project.kafka.consumer;

import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.wzh.project.common.Constants;

/**
 * 用于消费队列中的消息
 * @author wzh
 *
 */
public class KafkaConsumerThread implements Runnable{
	private static final Logger logger = LogManager.getLogger(KafkaConsumerThread.class);

	public static void main(String[] args) throws Exception {
		reciveMessage();
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		
	}
	
	public static void reciveMessage() throws Exception{
		Properties p = new Properties();
//		p.put("zookeeper.connect", "10.23.11.132:2181");
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.23.11.132:9092");
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
        kafkaConsumer.subscribe(Collections.singletonList(Constants.KAFKA_TOPIC));// 订阅消息

        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
            	logger.debug(String.format("topic:%s,offset:%d,消息:%s", //
                        record.topic(), record.offset(), record.value()));
            }
        }
	}
	
	
}
